package com.aotain.jupiter.flink;



import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import com.aotain.common.ZkKafkaConfig;
import com.aotain.jupiter.abnormal.AbnormalCheckFlink;
import com.aotain.jupiter.abnormal.AbnormalReduce;
import com.aotain.jupiter.abnormal.AbnormalStatMap;
import com.aotain.jupiter.abnormal.AbnormalStatTuple;
import com.aotain.jupiter.bots.BotsFlink;
import com.aotain.jupiter.bots.BotsReduce;
import com.aotain.jupiter.bots.BotsStatMap;
import com.aotain.jupiter.bots.BotsTuple;
import com.aotain.jupiter.cc.CCDetectFlink;
import com.aotain.jupiter.cc.URLStatMap;
import com.aotain.jupiter.cc.URLStatReduce;
import com.aotain.jupiter.common.URLStatTuple;
import com.aotain.jupiter.domain.DomainMap;
import com.aotain.jupiter.domain.DomainReduce;
import com.aotain.jupiter.domain.DomainStat;
import com.aotain.jupiter.flink.statis.AbnStatisMap;
import com.aotain.jupiter.flink.statis.AbnStatisTuple;
import com.aotain.jupiter.ipstat.IPFlowStat;
import com.aotain.jupiter.ipstat.IPFlowStatImport;
import com.aotain.jupiter.ipstat.IPFlowStatMap;
import com.aotain.jupiter.ipstat.IPStatTuple;
import com.aotain.jupiter.scanattack.ScanAttackFlink;
import com.aotain.jupiter.scanattack.ScanAttackReduce;
import com.aotain.jupiter.scanattack.ScanAttackStatMap;
import com.aotain.jupiter.scanattack.ScanAttackTuple;
import com.aotain.jupiter.sqlinjection.SQLInjectionDetect;
import com.aotain.jupiter.sqlinjection.SQLInjectionMap;
import com.aotain.jupiter.sqlinjection.SQLInjectionReduce;
import com.aotain.jupiter.sqlinjection.SQLInjectionTuple;
import com.aotain.jupiter.util.FilterNullEnum;
import com.aotain.jupiter.xss.XSSDetect;
import com.aotain.jupiter.xss.XSSMap;
import com.aotain.jupiter.xss.XSSReduce;
import com.aotain.jupiter.xss.XSSTuple;

public class SocketDataSourceFlink {

	public static void main(String[] args) throws Exception {

		String dbJson = "{\"driverClassName\":\"com.mysql.jdbc.Driver\","
				+ "\"maxActive\":\"200\",\"maxIdle\":\"50\",\"maxWait\":\"10000\","
				+ "\"name\":\"sds\",\"password\":\"h16aug8v3w\",\"queryTimeout\":\"60\","
				+ "\"type\":\"javax.sql.DataSource\",\"url\":\"jdbc:mysql://192.168.5.97:3306/SDS\""
				+ ",\"userName\":\"sds\""
				+ ",\"validationQuery\":\"SELECT SYSDATE()\"}";

		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		DataStream<Tuple2<String,String>> dataStream = env.socketTextStream("192.168.5.96", 9988)
				.map(new SourceDataMap()).setParallelism(8).map(new SourceDataStat(dbJson)).setParallelism(8).keyBy(0);

		//拆分需要统计的字段
		DataStream<Tuple2<String, IPStatTuple>> ipStatStream = dataStream.map(new IPFlowStatMap()); 

		WindowedStream<Tuple2<String, IPStatTuple>,Tuple,TimeWindow>  time5Min 
		= ipStatStream.keyBy(0).timeWindow(Time.minutes(5));//(TumblingEventTimeWindows.of(Time.minutes(5)));//(Time.minutes(5));
		time5Min.reduce(new IPFlowStat()).flatMap(new IPFlowStatImport("METIS_FLOW_STAT_5MIN"));

		WindowedStream<Tuple2<String, IPStatTuple>,Tuple,TimeWindow>  time30Min 
		= ipStatStream.keyBy(0).timeWindow(Time.minutes(30));//(TumblingEventTimeWindows.of(Time.minutes(5)));//(Time.minutes(5));
		time30Min.reduce(new IPFlowStat()).flatMap(new IPFlowStatImport("METIS_FLOW_STAT_30MIN"));

		WindowedStream<Tuple2<String, IPStatTuple>,Tuple,TimeWindow>  time60Min 
		= ipStatStream.keyBy(0).timeWindow(Time.minutes(60));//(TumblingEventTimeWindows.of(Time.minutes(5)));//(Time.minutes(5));
		time60Min.reduce(new IPFlowStat()).flatMap(new IPFlowStatImport("METIS_FLOW_STAT_HOUR"));


		//---------------------CC 攻击检测 ----START-----------
		WindowedStream<Tuple2<String, URLStatTuple>, Tuple, TimeWindow> winCC 
		= dataStream.map(new URLStatMap()).filter(new FilterNullObject<URLStatTuple>(FilterNullEnum.F1ISNOTNULL))
		.keyBy(0).timeWindow(Time.minutes(1));

		DataStream<AbnStatisTuple> abnStatCC = winCC.reduce(new URLStatReduce()).flatMap(new CCDetectFlink());
		//---------------------CC 攻击检测 ----END--------------

		ZkKafkaConfig z_kConfig = new ZkKafkaConfig("/home/storm/config/dbconfig.ini");

		//---------------------SQL 注入检测 ----START-----------
		WindowedStream<Tuple2<String, SQLInjectionTuple>, Tuple, TimeWindow> winSQLIn 
		= dataStream.map(new SQLInjectionMap(z_kConfig.getSQLInjection()))
		.filter(new FilterNullObject<SQLInjectionTuple>(FilterNullEnum.F1ISNOTNULL))
		.keyBy(0).timeWindow(Time.minutes(1));

		DataStream<AbnStatisTuple> abnStatSQLIn = 
				winSQLIn.reduce(new SQLInjectionReduce()).flatMap(new SQLInjectionDetect());
		//---------------------SQL 注入检测 ----END-----------

		//---------------------XSS 检测 ----START-----------
		WindowedStream<Tuple2<String, XSSTuple>, Tuple, TimeWindow> winXSS
		= dataStream.map(new XSSMap())
		.filter(new FilterNullObject<XSSTuple>(FilterNullEnum.F1ISNOTNULL))
		.keyBy(0).timeWindow(Time.minutes(1));

		//----得到异常传给异常统计Map
		DataStream<AbnStatisTuple> abnStatXSS = 
				winXSS.reduce(new XSSReduce()).flatMap(new XSSDetect());

		//---------------------XSS 注入检测 ----END-----------

		//---------------------DOMAIN 统计 ----START-----------
		WindowedStream<Tuple2<String, URLStatTuple>, Tuple, TimeWindow> winDomain 
		= dataStream.map(new DomainMap()).filter(new FilterNullObject<URLStatTuple>(FilterNullEnum.F1ISNOTNULL))
		.keyBy(0).timeWindow(Time.minutes(30));

		winDomain.reduce(new DomainReduce()).flatMap(new DomainStat());
		//---------------------DOMAIN 统计----END--------------


//		//--------------异常流量检测------------START-----/
//		WindowedStream<Tuple2<String,AbnormalStatTuple>,Tuple,TimeWindow> abnStream = dataStream.map(new AbnormalStatMap())
//				.filter(new FilterNullObject<AbnormalStatTuple>(FilterNullEnum.F1ISNOTNULL)).keyBy(0).timeWindow(Time.minutes(1));
//		abnStream.reduce(new AbnormalReduce()).flatMap(new AbnormalCheckFlink());
//		//--------------异常流量检测------------END-----/


		//--------------扫描攻击监测------------START---/
		WindowedStream<Tuple2<String,ScanAttackTuple>,Tuple,TimeWindow> scanAttStream = dataStream.map(new ScanAttackStatMap())
				.filter(new FilterNullObject<ScanAttackTuple>(FilterNullEnum.F0ISNOTNULL)).keyBy(0).timeWindow(Time.minutes(1));
		DataStream<AbnStatisTuple> scanAttStatis = scanAttStream.reduce(new ScanAttackReduce()).flatMap(new ScanAttackFlink());
		//--------------扫描攻击监测------------END-----/


		//--------------僵尸主机监测------------START---/
		WindowedStream<Tuple2<String,BotsTuple>,Tuple,TimeWindow> botsStream = dataStream.map(new BotsStatMap())
				.filter(new FilterNullObject<BotsTuple>(FilterNullEnum.F0ISNOTNULL)).keyBy(0).timeWindow(Time.minutes(1));
		DataStream<AbnStatisTuple> botsStatis = botsStream.reduce(new BotsReduce()).flatMap(new BotsFlink());
		//--------------僵尸主机监测------------END-----/

		//---------------------异常统计---------START--------------
		// ##scanattack
		WindowedStream<Tuple2<String,IPStatTuple>,Tuple,TimeWindow> scanAtt5MinWindow = scanAttStatis.map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(5));
		scanAtt5MinWindow.reduce(new IPFlowStat()).flatMap(new IPFlowStatImport("METIS_FLOW_STAT_5MIN"));
		WindowedStream<Tuple2<String,IPStatTuple>,Tuple,TimeWindow> scanAtt30MinWindow = scanAttStatis.map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(30));
		scanAtt30MinWindow.reduce(new IPFlowStat()).flatMap(new IPFlowStatImport("METIS_FLOW_STAT_30MIN"));
		WindowedStream<Tuple2<String,IPStatTuple>,Tuple,TimeWindow> scanAtt60MinWindow = scanAttStatis.map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(60));
		scanAtt60MinWindow.reduce(new IPFlowStat()).flatMap(new IPFlowStatImport("METIS_FLOW_STAT_HOUR"));
		
		// ##CC
		abnStatCC.map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(5))
			.reduce(new IPFlowStat()).flatMap(new IPFlowStatImport("METIS_FLOW_STAT_5MIN"));
		abnStatCC.map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(30))
			.reduce(new IPFlowStat()).flatMap(new IPFlowStatImport("METIS_FLOW_STAT_30MIN"));
		abnStatCC.map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(60))
			.reduce(new IPFlowStat()).flatMap(new IPFlowStatImport("METIS_FLOW_STAT_HOUR"));
		
		abnStatSQLIn.map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(5))
			.reduce(new IPFlowStat()).flatMap(new IPFlowStatImport("METIS_FLOW_STAT_5MIN"));
		abnStatSQLIn.map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(30))
			.reduce(new IPFlowStat()).flatMap(new IPFlowStatImport("METIS_FLOW_STAT_30MIN"));
		abnStatSQLIn.map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(60))
			.reduce(new IPFlowStat()).flatMap(new IPFlowStatImport("METIS_FLOW_STAT_HOUR"));
		
		abnStatXSS.map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(5))
			.reduce(new IPFlowStat()).flatMap(new IPFlowStatImport("METIS_FLOW_STAT_5MIN"));
		abnStatXSS.map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(30))
			.reduce(new IPFlowStat()).flatMap(new IPFlowStatImport("METIS_FLOW_STAT_30MIN"));
		abnStatXSS.map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(60))
			.reduce(new IPFlowStat()).flatMap(new IPFlowStatImport("METIS_FLOW_STAT_HOUR"));
//
//
//		// ##bots
//		WindowedStream<Tuple2<String,IPStatTuple>,Tuple,TimeWindow> bots5MinWindow = botsStatis.map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(5));
//		bots5MinWindow.reduce(new IPFlowStat()).flatMap(new IPFlowStatImport("METIS_FLOW_STAT_5MIN"));
//		WindowedStream<Tuple2<String,IPStatTuple>,Tuple,TimeWindow> bots30MinWindow = botsStatis.map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(30));
//		bots30MinWindow.reduce(new IPFlowStat()).flatMap(new IPFlowStatImport("METIS_FLOW_STAT_30MIN"));
//		WindowedStream<Tuple2<String,IPStatTuple>,Tuple,TimeWindow> bots60MinWindow = botsStatis.map(new AbnStatisMap()).keyBy(0).timeWindow(Time.minutes(60));
//		bots60MinWindow.reduce(new IPFlowStat()).flatMap(new IPFlowStatImport("METIS_FLOW_STAT_HOUR"));

		// ##abnormal


		//---------------------异常统计---------END--------------

		env.execute("SocketDataSourceFlink");
	}

}
